use async_trait::async_trait;
use std::sync::Mutex;
use common::traits::lf_message::LfMessage;
use crate::queue::queue_trait::EventQueue;

/// ZMQ 推送模式
#[derive(Clone, Copy, Debug)]
pub enum ZmqMode {
    Pub,
    Sub,
    Push,
    Pull,
}

/// ZMQ 队列实现：发送 TLV 二进制消息
pub struct ZmqQueue<T> {
    socket: Mutex<zmq::Socket>, // 线程安全保护
    _phantom: std::marker::PhantomData<T>,
    mode: ZmqMode
}

impl<T> ZmqQueue<T> {
    pub fn new(addr: &str, mode: ZmqMode) -> ZmqQueue<T> {
        let ctx = zmq::Context::new();
        let socket = match mode {
            ZmqMode::Pub => ctx.socket(zmq::PUB).unwrap(),
            ZmqMode::Push => ctx.socket(zmq::PUSH).unwrap(),
            ZmqMode::Sub => {
                let s = ctx.socket(zmq::SUB).unwrap();
                s.set_subscribe(b"").unwrap();
                s
            }
            ZmqMode::Pull => ctx.socket(zmq::PULL).unwrap(),
        };

        match mode {
            ZmqMode::Pub | ZmqMode::Push => {
                socket.bind(addr).unwrap();
            }
            ZmqMode::Sub | ZmqMode::Pull => {
                socket.connect(addr).unwrap();
            }
        }

        ZmqQueue {
            socket: std::sync::Mutex::new(socket),
            mode,
            _phantom: std::marker::PhantomData,
        }
    }
}

#[async_trait]
impl<T: LfMessage + Send + Sync + 'static> EventQueue<T> for ZmqQueue<T> {
    async fn enqueue(&self, item: T) {
        match self.mode {
            ZmqMode::Push | ZmqMode::Pub => {
                let bytes = item.encode_tlv();
                let socket = self.socket.lock().unwrap();
                let _ = socket.send(bytes, 0);
            }
            ZmqMode::Pull | ZmqMode::Sub => {
                panic!("enqueue not supported for ZmqMode::{:?}", self.mode);
            }
        }
    }

    async fn dequeue(&self) -> Option<T> {
        match self.mode {
            ZmqMode::Pull | ZmqMode::Sub => {
                let socket = self.socket.lock().unwrap();
                let msg = socket.recv_bytes(0).ok()?;
                T::decode_tlv(&msg).ok()
            }
            ZmqMode::Push | ZmqMode::Pub => {
                panic!("dequeue not supported for ZmqMode::{:?}", self.mode);
            }
        }
    }
}
